package com.bjsxt.kmeans

import scala.collection.mutable.ArrayBuffer
import scala.collection.mutable.ListBuffer

import org.apache.lucene.analysis.TokenStream
import org.apache.lucene.analysis.tokenattributes.CharTermAttribute
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.mllib.clustering.KMeans
import org.apache.spark.mllib.clustering.KMeansModel
import org.apache.spark.mllib.feature.HashingTF
import org.apache.spark.mllib.feature.IDF
import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.rdd.RDD
import org.wltea.analyzer.lucene.IKAnalyzer
import org.apache.spark.mllib.feature.IDFModel

object test {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("KMeans1").setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    val rdd = sc.textFile("./testdata.txt")
    /**
     * wordRDD 是一个KV格式的RDD
     * 	K:微博ID
     * 	V:微博内容分词后的结果 ArrayBuffer
     */
    var wordRDD = rdd.mapPartitions(iterator => {
      val list = new ListBuffer[(String, ArrayBuffer[String])]
      while (iterator.hasNext) {
        //创建分词对象   IKAnalyzer支持两种分词模式：最细粒度和智能分词模式，如果构造函数参数为false，那么使用最细粒度分词。
        val analyzer = new IKAnalyzer(true)
        val line = iterator.next()
        val textArr = line.split("\t")
        val id = textArr(0)
        val text = textArr(1)
        //分词     第一个参数只是标识性，没有实际作用，第二个读取的数据
        val ts : TokenStream = analyzer.tokenStream("", text)
        //得到相应词汇的内容
        val term : CharTermAttribute = ts.getAttribute(classOf[CharTermAttribute])
        //重置分词器，使得tokenstream可以重新返回各个分词
    		ts.reset()
    		val arr = new ArrayBuffer[String]
        //遍历分词数据
        while (ts.incrementToken()) {
          arr.+=(term.toString())
        }
        
        list.append((id, arr))
        analyzer.close()
      }
      list.iterator
    })
    wordRDD = wordRDD.cache()
    //九阳, 必须, 是, 其中之一, 的
    wordRDD.foreach(println)

    /**
     * HashingTF 使用hash表来存储分词
     * HashingTF 是一个Transformer 转换器，在文本处理中，接收词条的集合然后把这些集合转化成固定长度的特征向量,这个算法在哈希的同时会统计各个词条的词频
     * 1000:只是计算每篇微博中1000个单词的词频   最大似然估计思想
     */
    val hashingTF: HashingTF = new HashingTF(100)

    /**
     * tfRDD
     * K:微博ID
     * V:Vector（tf，tf，tf.....）
     * 
     * hashingTF.transform(x._2) 计算分词频数（TF）
     */
    val tfRDD = wordRDD.map(x => {
      (x._1, hashingTF.transform(x._2))
    })
    tfRDD.foreach(println)

    /**
     * 得到IDFModel，要计算每个单词在整个语料库中的IDF
     * IDF是一个 Estimator 评价器，在一个数据集上应用它的fit（）方法，产生一个IDFModel。 该IDFModel 接收特征向量（由HashingTF产生）
     * new IDF().fit(tfRDD.map(_._2)) 就是在组织训练这个评价器，让评价器知道语料库中有那些个词块，方便计算IDF
     */
    val idf: IDFModel = new IDF().fit(tfRDD.map(_._2))

    /**
     * K:微博 ID
     * V:每一个单词的TF-IDF值
     * tfIdfs这个RDD中的Vector就是训练模型的训练集
     */
    val tfIdfs: RDD[(String, Vector)] = tfRDD.mapValues(idf.transform(_))
    tfIdfs.foreach(x=>{
        println("tfIdfs = "+x)
    })
    //设置聚类个数
    val kcluster = 20
    val kmeans = new KMeans()
    kmeans.setK(kcluster)
    //使用的是kemans++算法来训练模型  "random"|"k-means||"
    kmeans.setInitializationMode("k-means||")
    //设置最大迭代次数
    kmeans.setMaxIterations(100)
    //训练模型
    val kmeansModel: KMeansModel= kmeans.run(tfIdfs.map(_._2))
    //    kmeansModel.save(sc, "d:/model001")
    //打印模型的20个中心点
    val centers = kmeansModel.clusterCenters
    centers.foreach(println)
//    println(kmeansModel.clusterCenters)

    /**
     * 模型预测
     */
    val modelBroadcast = sc.broadcast(kmeansModel)
    /**
     * predicetionRDD KV格式的RDD
     * 	K：微博ID
     * 	V：分类号
     */
    val predicetionRDD = tfIdfs.mapValues(sample => {
      val model = modelBroadcast.value
      model.predict(sample)
    })
//    predicetionRDD.saveAsTextFile("d:/resultttt")

    /**
     * 总结预测结果
     * tfIdfs2wordsRDD:kv格式的RDD
     * K：微博ID
     * V：二元组(Vector(tfidf1,tfidf2....),ArrayBuffer(word,word,word....))
     */
    val tfIdfs2wordsRDD = tfIdfs.join(wordRDD)
    /**
     * result:KV
     * K：微博ID
     * V:(类别号，(Vector(tfidf1,tfidf2....),ArrayBuffer(word,word,word....)))
     */
    val result = predicetionRDD.join(tfIdfs2wordsRDD)

    /**
     * 查看0号类别中tf-idf比较高的单词，能代表这类的主题
     */
    result
      .filter(x => x._2._1 == 0)
      .flatMap(line => {
        
        val tfIdfV: Vector = line._2._2._1
        val words: ArrayBuffer[String] = line._2._2._2
        val tfIdfA: Array[Double] = tfIdfV.toArray
        
        val wordL = new ListBuffer[String]()
        val tfIdfL = new ListBuffer[Double]()
        var index = 0
        for(i <- 0 until tfIdfA.length ;if tfIdfV(i) != 0){
          wordL.+=(words(index))
          tfIdfL.+=(tfIdfA(index))
          index += 1
        }
        println(wordL.length + "===" + tfIdfL.length)
        val list = new ListBuffer[(Double, String)]
        for (i <- 0 until wordL.length) {
          list.append((tfIdfV(i), words(i)))
        }
        list
      }).map(_.swap).reduceByKey(_+_).map(_.swap)
      .sortBy(x => x._1, false)
      .map(_._2)
      .take(30).foreach(println)
    sc.stop()
  }

}